package com.zf.flink;// com/example/source/MySQLSource.java
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DriverManagerDataSource;

import java.util.List;
import java.util.concurrent.TimeUnit;

public class MySQLSource extends RichSourceFunction<User> {

    private volatile boolean isRunning = true;
    private transient JdbcTemplate jdbcTemplate;

    @Override
    public void open(Configuration parameters) {
        DriverManagerDataSource dataSource = new DriverManagerDataSource();
        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
        dataSource.setUrl("jdbc:mysql://localhost:3306/bobdb?useSSL=false&serverTimezone=UTC");
        dataSource.setUsername("root");
        dataSource.setPassword("123456");

        this.jdbcTemplate = new JdbcTemplate(dataSource);
    }

    @Override
    public void run(SourceContext<User> ctx) throws Exception {
        while (isRunning) {
            List<User> users = jdbcTemplate.query(
                "SELECT id, name, age FROM users",
                (rs, rowNum) -> {
                    User user = new User();
                    user.setId(rs.getLong("id"));
                    user.setName(rs.getString("name"));
                    user.setAge(rs.getInt("age"));
                    return user;
                }
            );

            for (User user : users) {
                ctx.collect(user);
            }

            TimeUnit.SECONDS.sleep(5); // 模拟定时拉取
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}